package com.com.zs21cp.ApiDemo;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;


public class Connet_Unino {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
        DataStreamSource<String> source2 = env.fromElements("1", "2", "3", "4", "5", "6", "7");
        DataStreamSource<String> source3 = env.fromElements("31", "21", "31", "41", "51", "16", "71");
        DataStreamSource<String> source4 = env.fromElements("23", "24", "32", "45", "51", "64", "72");
        //unino只能连接相同的流，类型必须一样
        DataStream<String> union = source2.union(source3).union(source4);
        //connect 只能链接两条流，类型可以不一样，且不能直接输出，在后面跟上map或者flatmap将算子转换以后才可以
        ConnectedStreams<Integer, String> connect = source1.connect(source2);
        union.print();
        //map 和flatmap的区别：当我们需要将具有层级结构的数据展平时，也就是将多层数据转换为单层数据操作时，
        // 我们可以使用 flatMap 方法。如果我们只是简单的对流中的数据计算或者转换时，可以使用 map 方法。
        // 比如说 使用 ①latMap: [ a，b，c，d，[ e，f [ g，h，i ]]转换为[ a，b，c，d，e，f，g，h，i ]
        //           ② 使用 map: [1，2，3，4，5，6] 转换为 [11，12，13，14，15，16]
        connect.map(new CoMapFunction<Integer, String, String>() {
            @Override
            public String map1(Integer value) throws Exception {

                return null;
            }

            @Override
            public String map2(String value) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {

                }
                return null;
            }
        });
//        //conect可以对两条流做分开处理
//        connect.flatMap(new CoFlatMapFunction<Integer, String, String>() {
//            @Override
//            public void flatMap1(Integer value, Collector<String> out) throws Exception {
//                out.collect(value.toString());
//            }
//
//            @Override
//            public void flatMap2(String value, Collector<String> out) throws Exception {
//                out.collect(value);
//            }
//        }).print();
        env.execute();

    }
}
